package cn.virens.server.aio;

import java.io.Closeable;
import java.io.IOException;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.hutool.core.collection.ConcurrentHashSet;
import cn.virens.common.fun.ExConsumer;
import cn.virens.common.utils.common.CloseableUtil;
import cn.virens.server.aio.channel.AioChannel;
import cn.virens.server.aio.channel.AioChannelSimple;
import cn.virens.server.aio.handler.PacketHandler;
import cn.virens.server.aio.listener.AioDataDecoder;
import cn.virens.server.aio.listener.AioDataEncoder;
import cn.virens.server.aio.listener.AioDataErrorListener;
import cn.virens.server.aio.listener.AioDataReadListener;
import cn.virens.server.aio.listener.AioDataWriteListener;
import cn.virens.server.aio.listener.AioStateCloseableListener;

public class AsynchronousSocketAcceptHandler implements CompletionHandler<AsynchronousSocketChannel, Object>, Closeable {
	private final Logger logger = LoggerFactory.getLogger(AsynchronousSocketAcceptHandler.class);
	private final Map<String, AioChannel> channelMap = new ConcurrentHashMap<>();
	private final Set<AioChannel> channelSet = new ConcurrentHashSet<>();

	private final AioServerManageImpl serverManage;

	private AsynchronousSocketAcceptHandler(AioServerManageImpl serverManage) {
		this.serverManage = serverManage;
	}

	@Override
	public void completed(AsynchronousSocketChannel asynchronousSocketChannel, Object attachment) {
		AioChannelSimple channel = null;

		try {
			channel = new AioChannelSimple(asynchronousSocketChannel, this);

			// 判断连接是否已经打开，如果连接打开就将连接放入连接池
			if (!serverManage.isWaitStop() && channel.isOpen() && channelSet.add(channel)) {
				channel.setOption(StandardSocketOptions.SO_RCVBUF, 64 * 1024);
				channel.setOption(StandardSocketOptions.SO_SNDBUF, 64 * 1024);
				channel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
				channel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
				channel.start(ByteBuffer.allocate(2048));

				// 连接已经完成创建，回调连接成功事件
				if (serverManage.aioStateConnectedListener != null) {
					serverManage.aioStateConnectedListener.onConnected(channel);
				}
			} else {
				CloseableUtil.close(asynchronousSocketChannel);
			}
		} catch (Exception ex) {
			// 从连接池中删掉该连接并安全的关闭该连接
			if (channel == null || !channelSet.remove(channel)) {
				CloseableUtil.close(asynchronousSocketChannel);
			} else {
				CloseableUtil.close(channel);
			}

			// 如果有异常回调监听，就使用异常和连接信息进行回调
			if (channel != null && serverManage.aioStateExceptionListener != null) {
				serverManage.aioStateExceptionListener.onFailed(channel, ex);
			}
		} finally {
			// 如果还没开始关闭服务，就继续进行等待连接
			if (serverManage.serverSocketChannel.isOpen() && !serverManage.isWaitStop()) {
				serverManage.serverSocketChannel.accept(null, this);
			}
		}
	}

	@Override
	public void failed(Throwable exc, Object attachment) {
		if (exc instanceof AsynchronousCloseException) {
			logger.debug("停止监听连接进入，并关闭所有的线程池...");

			this.ex((serverManage) -> serverManage.channelGroup.shutdownNow());
			this.ex((serverManage) -> serverManage.groupExecutor.shutdownNow());
			this.ex((serverManage) -> serverManage.wokerExecutor.shutdownNow());
		} else if (serverManage.aioStateExceptionListener != null) {
			serverManage.aioStateExceptionListener.onFailed(null, exc);
		}
	}

	public AioChannel getChannel(String uid) {
		return channelMap.get(uid);
	}

	public Set<AioChannel> getChannelAll() {
		return new HashSet<>(channelSet);
	}

	public Set<String> getChannelKeys() {
		return new HashSet<>(channelMap.keySet());
	}

	public void bindUid(AioChannel aioChannel, String uid) {
		if (aioChannel instanceof AioChannelSimple) {
			((AioChannelSimple) aioChannel).setRemove(false);
			((AioChannelSimple) aioChannel).setUid(uid);
		}

		// 将用户ID和连接关联起来
		this.channelMap.put(uid, aioChannel);
	}

	public void unbindUid(AioChannel aioChannel) {
		ex((serverManage) -> ((AioChannelSimple) aioChannel).setRemove(true));
		ex((serverManage) -> ((AioChannelSimple) aioChannel).close());

		// 如果连接已经绑定了uid值，就需要释放uid绑定关系
		if (aioChannel != null && aioChannel.getUid() != null) {
			this.channelMap.remove(aioChannel.getUid());
		}

		// 从连接池中释放连接,如果是第一次调用并且启动时注册了连接关闭监听，就调用关闭监听
		if (channelSet.remove(aioChannel) && serverManage.aioStateCloseableListener != null) {
			serverManage.aioStateCloseableListener.onClose(aioChannel);
		}
	}

	public void execute(Runnable task) {
		this.serverManage.wokerExecutor.execute(task);
	}

	public static AsynchronousSocketAcceptHandler newInstance(AioServerManageImpl serverManage) {
		return new AsynchronousSocketAcceptHandler(serverManage);
	}

	public List<PacketHandler> getHandler() {
		return serverManage.handlerPool;
	}

	public AioDataDecoder getAioDecoder() {
		return serverManage.aioDataDecoder;
	}

	public AioDataEncoder getAioEncoder() {
		return serverManage.aioDataEncoder;
	}

	public AioDataReadListener getAioDataReadListener() {
		return serverManage.aioDataReadListener;
	}

	public AioDataWriteListener getAioDataWriteListener() {
		return serverManage.aioDataWriteListener;
	}

	public AioDataErrorListener getAioDataErrorListener() {
		return serverManage.aioDataErrorListener;
	}

	public AioStateCloseableListener getAioStateCloseableListener() {
		return serverManage.aioStateCloseableListener;
	}

	public void close() throws IOException {
		for (AioChannel aioChannel : channelSet) {
			ex((serverManage) -> aioChannel.close());
		}
	}

	private void ex(ExConsumer<AioServerManageImpl> consumer) {
		try {
			consumer.accept(serverManage);
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}

}
